/**
 * @author lijing
 * @since 2023-01-03
 * 益丰自研全量获取数据Function
 */

package com.yifeng.repo.flink.data.transport.streaming.functions;

import java.io.IOException;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.dto.DataPacket;
import com.yifeng.repo.flink.data.transport.streaming.api.Validator;
import com.yifeng.repo.flink.data.transport.streaming.functions.api.Engine;
import com.yifeng.repo.flink.data.transport.streaming.functions.task.FullSyncTaskFactory;

/**
 * 全量数据同步function
 * @author lijing
 * @since 2023-01-03
 *
 * @param <T>
 */
public class FullSourceFunction<T> extends RichSourceFunction<DataPacket> {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(FullSourceFunction.class);

	private static final long serialVersionUID = 1L;

	//全量同步前校验
    private final Validator validator;
    
    private final DataSourceConfig dataSourceConfig;
    
    private transient Engine engine;
    
	public FullSourceFunction(DataSourceConfig dataSourceConfig, Validator validator) {
        this.validator = validator;
        this.dataSourceConfig = dataSourceConfig;
    }
	
	@Override
    public void open(Configuration parameters) throws Exception {
        validator.validate();
        super.open(parameters);
	}
	
	@Override
	public void close() throws Exception {
		cancel();
		super.close();
		LOGGER.info("同步全量任务结束");
	}
	
	/**
	 * 最终执行异步的方法，采用流式读取数据库中的数据
	 * */
	@Override
	public void run(SourceContext<DataPacket> ctx) throws Exception {
		LOGGER.info("开始进行全量数据采集");
		this.engine = new FullSyncTaskFactory().builder(dataSourceConfig);
		engine.run(ctx);
	}
	
	@Override
	public void cancel() {
		shutdownEngine();
	}
	
	/** Safely and gracefully stop the full engine. */
    private void shutdownEngine() {
        try {
            if (engine != null) {
                engine.close();
            }
        } catch (IOException e) {
            ExceptionUtils.rethrow(e);
        }
    }

}
